package main

import (
	"common/process"
	"flag"
	"iot-bridge/iot_cmd_up/distribute/conf"
	"iot-bridge/iot_cmd_up/distribute/controller"
	"iot-bridge/iot_cmd_up/distribute/dao"
	"iot-bridge/iot_cmd_up/distribute/service"
	"os"
	"os/signal"
	"runtime"
	"sync"
	"syscall"
	"time"

	log "common/log4go.v1"
)

var (
	confPath string
)

func init() {
	flag.StringVar(&confPath, "conf", "./conf_ts.ini", " set config file path")
}

func main() {
	flag.Parse()
	if err := conf.Init(confPath); err != nil {
		log.Error("conf.Init(\"%s\") error(%v)", confPath, err)
		panic(err)
	}
	runtime.GOMAXPROCS(conf.Conf.Base.MaxProc)
	log.LoadConfiguration(conf.Conf.Base.LogPath)
	defer func() {
		log.Close()
	}()
	defer dao.Close()
	if err := dao.Init(); err != nil {
		log.Error("dao.Init() error(%v)", err)
		return
	}

	if err := process.Init(conf.Conf.Base.PidFile); err != nil {
		log.Error("process.Init(%v) error(%v)", conf.Conf.Base.PidFile, err)
		return
	}
	defer service.Close()
	if err := service.Init(); err != nil {
		log.Error("service.Init(%v) error(%v)", err)
		return
	}
	defer controller.Close()
	if err := controller.Init(); err != nil {
		log.Error("controller.Init(%v) error(%v)", err)
		return
	}
	var wg = &sync.WaitGroup{}
	//消费者
	//for i := 0; i < len(conf.Conf.Kafka.BrokerList); i++ {
	//wg.Add(1)
	go controller.TcpClusterConsumer(wg, conf.Conf.Kafka.ZKList, conf.Conf.Kafka.BrokerList, conf.Conf.Kafka.GroupId)
	//}
	//wg.Wait()

	log.Info("TcpKafkaConsumer start")
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
	for {
		s := <-c
		log.Info("get a signal %s", s.String())
		switch s {
		case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:

			log.Info("TcpKafkaConsumer exit %v", os.Getpid())
			time.Sleep(time.Second)
			return
		case syscall.SIGHUP:
		// TODO reload
		default:
			return
		}
	}
}
